package com.winning.bi.job;

import java.io.IOException;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import org.apache.commons.lang3.StringUtils;
import org.bson.Document;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.PersistJobDataAfterExecution;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.winning.bi.amqp.SendMessageUtils;
import com.winning.bi.data.Message;
import com.winning.bi.jdbc.QuartzUtils;
import com.winning.bi.jdbc.ScheduleJob;
import com.winning.bi.jdbc.dao.Dao;

import mongo.MongoDbUtil;

@DisallowConcurrentExecution
@PersistJobDataAfterExecution

/**
 * 老版本分页，分页是为了防止数据量过大 使用amqp协议
 * 
 * @author 张大川
 *
 */

public class OldSqlJob implements Job {
	private static Logger log = LoggerFactory.getLogger(OldSqlJob.class);
	Map<String, Object> map2;
	static long page = 200;
	String sql;

	/**
	 * 转换格式
	 * 
	 * @param columnMap
	 * @return
	 */

	void saveToMongo(List<Map<String, Object>> list, String taskName) {

		for (int i = 0; i < list.size(); i++) {
			Document documentDetail = new Document();
			documentDetail.put("loadtime", LocalDateTime.now().toString());
			documentDetail.put("content", list.get(i));
			documentDetail.put("taskname", taskName);
			MongoDbUtil.getDatabase("test").getCollection("log").insertOne(documentDetail);
		}

	}

	public void execute(JobExecutionContext context) {

		ScheduleJob job = (ScheduleJob) context.getMergedJobDataMap().get("scheduleJob");

		Map<String, Object> properties = job.getMap();
		ClassPathXmlApplicationContext ac = (ClassPathXmlApplicationContext) context.getMergedJobDataMap().get("ac");
		Dao dao = (Dao) ac.getBean("daoSource");
		log.info("[任务名]" + properties.get("jobname"));

		String rowversion = context.getJobDetail().getJobDataMap().get("rowversion").toString();
		if (StringUtils.isBlank(rowversion)) {
			//
			rowversion = "0x0000000000000000";
		}

		log.info("[处理里程碑]" + rowversion);

		byte[] bs = dao.getMaxVersion(properties.get("sourcetable").toString());
		boolean query = false;
		String now = QuartzUtils.bytesToHexString(bs);
		log.info("[现在]" + now);

		Timestamp timestamp = new Timestamp(System.currentTimeMillis());
		String updateSql = "update task_info set markversion = ? , processtime = ? where jobid=?";

		dao.updatetaskInfo(updateSql, now, timestamp, properties.get("jobid"));
		context.getJobDetail().getJobDataMap().put("rowversion", now);
		String countSql = "Select count(*) from " + properties.get("sourcetable") + " WHERE  timestamp > " + rowversion
				+ " AND  timestamp <=" + now;
		log.info("[countSql]" + countSql);
		Map<String, Object> logMap = new HashMap<String, Object>();

		logMap.put("taskid", properties.get("jobid"));
		logMap.put("taskname", properties.get("jobname"));
		logMap.put("beginstamp", rowversion);
		logMap.put("endstamp", now);
		logMap.put("executetime", timestamp);
		if (StringUtils.compareIgnoreCase(now, rowversion) > 0) {
			log.info("有数据更新");
			query = true;
		} else {
			log.info("无数据更新");
			logMap.put("amount", 0L);
		}
		if (query) {
			String mapString = properties.get("mapping").toString();
			List<String> columList = new ArrayList<String>();// 字段集合
			Map<String, String> ll = JSON.parseObject(mapString, new TypeReference<Map<String, String>>() {
			});

			for (Map.Entry<String, String> entry : ll.entrySet()) {
				columList.add(entry.getKey());
			}

			String from = StringUtils.remove(columList.toString(), "[");
			String string = StringUtils.remove(from, "]");
			if (!ll.containsKey("timestamp")) {//如果配置的映射中不包换时间戳，则添加伊特
				string = string + ", timestamp";
			}

			Long long1 = dao.getCountNumber(countSql);
			System.out.println("总数:" + long1);
			logMap.put("amount", long1);

			String sqltemp = "SELECT " + string + " FROM    ( SELECT TOP _pagecount "
					+ " ROW_NUMBER() OVER ( ORDER BY timestamp  ) AS rownum , * from " + properties.get("sourcetable")
					+ " WHERE  timestamp >" + rowversion + " AND timestamp<=" + now
					+ " ) AS temp WHERE   temp.rownum > _pageno ORDER BY timestamp";
			Long pageCount = long1 / page + 1;
			log.info("总页数" + pageCount);
			SendMessageUtils messageUtils = null;
			try {
				messageUtils = new SendMessageUtils();
			} catch (IOException | TimeoutException e1) {
				
				e1.printStackTrace();
			}
			Map<String, String> map = dao.getMetaData(properties.get("sourcetable").toString());
			for (Long i = 1L; i < pageCount + 1; i++) {
				log.info("第几页:" + i);
				sql = StringUtils.replace(sqltemp, "_pagecount ", i * page + "");
				sql = StringUtils.replace(sql, "_pageno", page * (i - 1) + "");
				log.info("sql:" + sql);

				List<Map<String, Object>> list = null;

				list = dao.getListMap(sql, map, ll);
				for (int j = 0; j < list.size(); j++) {
					Message message = new Message();
					map2 = list.get(j);
					try {
						map2.remove("timestamp");

					} catch (Exception e) {
						log.debug("不包含时间戳");
					}
					message.setMap(map2);
					message.setTarget(properties.get("targettable").toString());
					log.info("[message]" + JSON.toJSONString(message));
					try {

						messageUtils.sendMessage(properties.get("queuename").toString(), message);
					} catch (Exception e) {

						e.printStackTrace();
					}
				}
				// try {
				// messageUtils.close();
				// } catch (IOException e) {
				// // TODO Auto-generated catch block
				// e.printStackTrace();
				// }
			}

		}

		try {
			dao.insertMap(logMap, "task_log");
		} catch (SQLException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

	}

}
